// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#pragma once

#include <gen_cpp/parquet_types.h>

#include "common/status.h"
#include "vec/data_types/data_type.h"
#include "vec/exec/format/parquet/decoder.h"
#include "vec/exec/format/parquet/parquet_common.h"

namespace doris::vectorized {
class ColumnSelectVector;
} // namespace doris::vectorized

namespace doris::vectorized {
#include "common/compile_check_begin.h"

class FixLengthPlainDecoder final : public Decoder {
public:
    FixLengthPlainDecoder() = default;
    ~FixLengthPlainDecoder() override = default;

    Status decode_values(MutableColumnPtr& doris_column, DataTypePtr& data_type,
                         ColumnSelectVector& select_vector, bool is_dict_filter) override;

    template <bool has_filter>
    Status _decode_values(MutableColumnPtr& doris_column, DataTypePtr& data_type,
                          ColumnSelectVector& select_vector, bool is_dict_filter) {
        size_t non_null_size = select_vector.num_values() - select_vector.num_nulls();
        if (UNLIKELY(_offset + _type_length * non_null_size > _data->size)) {
            return Status::IOError("Out-of-bounds access in parquet data decoder");
        }

        size_t primitive_length = remove_nullable(data_type)->get_size_of_value_in_memory();
        size_t data_index = doris_column->size() * primitive_length;
        size_t scale_size = (select_vector.num_values() - select_vector.num_filtered()) *
                            (_type_length / primitive_length);
        doris_column->resize(doris_column->size() + scale_size);
        // doris_column is of type MutableColumnPtr, which uses get_raw_data
        // to return a StringRef, hence the use of const_cast.
        char* raw_data = const_cast<char*>(doris_column->get_raw_data().data);
        ColumnSelectVector::DataReadType read_type;
        while (size_t run_length = select_vector.get_next_run<has_filter>(&read_type)) {
            switch (read_type) {
            case ColumnSelectVector::CONTENT: {
                memcpy(raw_data + data_index, _data->data + _offset, run_length * _type_length);
                _offset += run_length * _type_length;
                data_index += run_length * _type_length;
                break;
            }
            case ColumnSelectVector::NULL_DATA: {
                data_index += run_length * _type_length;
                break;
            }
            case ColumnSelectVector::FILTERED_CONTENT: {
                _offset += _type_length * run_length;
                break;
            }
            case ColumnSelectVector::FILTERED_NULL: {
                // do nothing
                break;
            }
            }
        }
        return Status::OK();
    }

    Status skip_values(size_t num_values) override;
};

#include "common/compile_check_end.h"

} // namespace doris::vectorized
